package test.mq;

import static org.junit.Assert.*;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import javax.jms.JMSException;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import hyl.base.mq.kafka.MyKafka;
import hyl.core.MyDate;
import hyl.core.run.IDo;
import hyl.core.run.IRun;
import hyl.core.run.MyRun;

public class testkafka {
	MyKafka kfkp = MyKafka.getInsProducer();
	MyKafka kfks = MyKafka.getInsConsumer("");

	public static void main(String[] args) throws Exception {
		testkafka tk = new testkafka();
		// tk. test();
		tk.test2();
	}

	// @Test
	public void test() {
		MyRun mr = MyRun.getInstance("测试", 12);

		mr.execute(new Runnable() {
			@Override
			public void run() {
				try {
					subsyn1();
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});

		class ps implements Runnable {
			int k = 0;

			ps(int i) {
				k = i;
			}

			@Override
			public void run() {
				pubsyn1(k);
			}
		}
		for (int i = 0; i < 10; i++) {
			mr.execute(new ps(i));
		}

		System.out.println("结束....");
		try {
			Thread.currentThread().join();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		kfks.close();
		mr.close();
		// mr.waitAndClose();
	}

	// @Before
	public void before() {
	}

	public void pubsyn1(int i) {

		kfkp.sendSyn("testsyn", ("键" + i).getBytes(), ("值" + i).getBytes());
		System.out.println("发送" + i);
		MyDate.sleep(1000);

	}
	
	// @Test
	public void test2() throws InterruptedException, ExecutionException {
		MyRun mr = MyRun.getInstance("收线程", 1);
		MyRun mr2 = MyRun.getInstance("发线程", 10);

		class ps2 implements Callable<String> {
			int k = 0;

			ps2(int i) {
				k = i;
			}

			@Override
			public String call() throws Exception {
				pubasyn1(k);
				return String.valueOf(k);
			}
		}

		Future<Object> fu = mr.submit(new Callable<Object>() {
			@Override
			public String call() throws Exception {
				subasyn1();
				return "";
			}
		});

		// fu.get();
		for (int i = 0; i < 9; i++) {
			System.out.println("i=" + i);
			mr2.submit(new ps2(i));

		}

		Thread.currentThread().join();
		System.out.println("结束....");

		kfks.close();
		MyRun.closeAll();
	}

	public void pubasyn1(int i) {

		kfkp.send("testasyn", ("键" + i).getBytes(), ("值" + i).getBytes());
		System.out.println("发送=" + i);
		// MyDate.sleep(1000);
		try {
			Thread.currentThread().sleep(1000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	@Test
	public void subasyn1() {
		System.out.println("开始异步接收.....");
		IRun d = new IRun() {
			@SuppressWarnings("unchecked")
			@Override
			public void run(Object obj) {
				System.out.println("--------------------------------------------------");
				ConsumerRecord<byte[], byte[]> cr = (ConsumerRecord<byte[], byte[]>) obj;
				System.out.println("收到=" + new String(cr.key()) + ',' + new String(cr.value()));
				kfks.commit();
			}
		};
		kfks.receive("testasyn", d);
	}
	//@Test
	public void subsyn1() throws Exception {
		System.out.println("开启同步接收.....");
		IRun d = new IRun() {
			@SuppressWarnings("unchecked")
			@Override
			public void run(Object obj) {
			   System.out.println("------------------");

				ConsumerRecord<byte[], byte[]> cr = (ConsumerRecord<byte[], byte[]>) obj;
				System.out.println("收到=" + new String(cr.key()) + ',' + new String(cr.value()));
				kfks.commitSyn();
			}
		};
		kfks.receiveSyn("testsyn", d);
	
	}

	@After
	public void after() {
		kfks.close();
		kfkp.close();
		System.out.println("关闭");
	}
}
